package com.yl.flink.processor;

import cn.hutool.core.util.StrUtil;
import com.yl.constant.Const;
import com.yl.entity.MultiDataEntity;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.connectors.influxdb.InfluxDBPoint;

import java.util.HashMap;
import java.util.Map;

/**
 * @author wlf
 * @since 2022/8/18
 */
public class CalPointMapFunc implements MapFunction<MultiDataEntity, InfluxDBPoint> {

    /**
     * @param multiDataEntity 流元素
     */
    @Override
    public InfluxDBPoint map(MultiDataEntity multiDataEntity) throws Exception {
        // 表名
        String tbl = StrUtil.format(
                Const.TPL_TBL_CAL,
                // 以测点类型tag为表名前缀
                multiDataEntity.getTargetType().getTag().toLowerCase());
        // 数据时间
        long timestamp = multiDataEntity.getPickTime();
        // tags
        HashMap<String, String> tags = new HashMap<>();
        tags.put(Const.S_TID, multiDataEntity.getTarget().getId().toString());
        tags.put(Const.S_CALK, multiDataEntity.getCalKey());
        // fields
        HashMap<String, Object> fields = new HashMap<>();
        Tuple2<Integer, Double> calValues = multiDataEntity.getCalValues();
        fields.put(StrUtil.format(Const.TPL_FIELD, calValues.f0.toString()), calValues.f1);
        return new InfluxDBPoint(tbl, timestamp, tags, fields);
    }
}
